package com.corn.flink.lesson6;

import cn.hutool.core.collection.CollUtil;
import com.corn.flink.lesson4.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * @author JimWu
 * @date 2023/3/23 17:17
 **/
public class KeyedStreamTimerAndTimeServiceDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        List<String> eventType = CollUtil.newArrayList("add", "delete", "update", "select");
        DataStreamSource<Event> ds = env.addSource(new RichSourceFunction<Event>() {
            private boolean flag = true;

            @Override
            public void run(SourceContext<Event> ctx) throws Exception {
                while (flag) {
                    ctx.collect(new Event(
                            eventType.get(ThreadLocalRandom.current().nextInt(eventType.size())),
                            System.currentTimeMillis()
                    ));
                    TimeUnit.SECONDS.sleep(1);
                }
            }

            @Override
            public void cancel() {
                this.flag = false;
            }
        });
        ds.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }))
                .keyBy(data -> true)
                .process(new KeyedProcessFunction<Boolean, Event, String>() {
                             @Override
                             public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {

                                 // 获取timeService
                                 TimerService timerService = ctx.timerService();
                                 // 注册一个2s之后的事件定时器
//                                 timerService.registerEventTimeTimer(timerService.currentWatermark() + 2 * 1000);
                                 // 注册5S之后的处理定时器
                                 timerService.registerProcessingTimeTimer(timerService.currentProcessingTime() + 5 * 1000);
                                 out.collect(value.type + " -> " + new Timestamp(value.timestamp));
                             }

                             @Override
                             public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                                 out.collect("当前定时器触发: -> " + new Timestamp(timestamp));
                             }
                         }


                )
                .print();

        env.execute();

    }
}
